package xiegd.study.kafka.springboot;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import kafka.utils.json.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Date;
import java.util.UUID;

/**
 * @Description:
 * @Author: xieguangda
 * @Date: 2021年06月30日 15:08
 */
@Component
@Slf4j
public class KafkaSender {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    //发送消息方法
    public void send() throws Exception{
        Message message = new Message();
        message.setId(System.currentTimeMillis());
        message.setMsg(UUID.randomUUID().toString());
        message.setSendTime(new Date());
      //  log.info("+++++++++++++++++++++  message = {}", JSONObject.toJSONString(message));
        System.out.println("+++++++++++++++++++++  message " + JSONObject.toJSONString(message));
        kafkaTemplate.send("kafkastudy", JSONObject.toJSONString(message)).get();
//        ListenableFuture<SendResult<String, String>> kafkastudy6 = kafkaTemplate.send("kafkastudy", JSONObject.toJSONString(message));
//        kafkastudy6.addCallback(new ListenableFutureCallback<SendResult<String, String >>() {
//            @Override
//            public void onSuccess(SendResult<String, String> stringStringSendResult) {
//                System.out.println("kafka消息推送成功：" + message);
//            }
//
//            @Override
//            public void onFailure(Throwable throwable) {
//                System.out.println("kafka消息推送成功：" + throwable.getMessage());
//            }
//        });
    }

    //延迟队列，不是延迟返送，是发送过去了，但是没有消费
    public void sendAndDelayConsumer() throws  Exception{
        Message message = new Message();
        message.setId(System.currentTimeMillis());
        message.setMsg(UUID.randomUUID().toString());
        message.setSendTime(new Date());
        //  log.info("+++++++++++++++++++++  message = {}", JSONObject.toJSONString(message));
        System.out.println("+++++++++++++++++++++  message " + JSONObject.toJSONString(message));
        kafkaTemplate.send("kafkadelaystudy", JSONObject.toJSONString(message)).get();
    }
}